package com.atguigu.chapter07;

import com.atguigu.chapter02.Event;
import com.atguigu.chapter05.Clicksource;
import com.atguigu.chapter06.UrlViewCount;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;

import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.*;

public class ProcessAllWindowTopN {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> stream = env.addSource(new Clicksource()).
                assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO).
                        withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override
                            public long extractTimestamp(Event event, long l) {
                                return event.timestamp;
                            }
                        }));

        stream.map(a -> a.url).windowAll(SlidingEventTimeWindows.of(Time.seconds(10),Time.seconds(5))).
                aggregate(new UrlViewCountAgg(), new  UrlViewCountResult()).print();


        env.execute();
    }


    private static class UrlViewCountAgg implements AggregateFunction<String, HashMap<String,Long>,ArrayList<Tuple2<String,Long>> > {


        @Override
        public HashMap<String, Long> createAccumulator() {
            return new HashMap<>();
        }

        @Override
        public HashMap<String, Long> add(String s, HashMap<String, Long> stringLongHashMap) {
            if(stringLongHashMap.containsKey(s)==true){
                stringLongHashMap.put(s,stringLongHashMap.get(s)+1);
            }else {
                stringLongHashMap.put(s,1L);
            }

            return stringLongHashMap;
        }

        @Override
        public ArrayList<Tuple2<String, Long>> getResult(HashMap<String, Long> stringLongHashMap) {
            ArrayList<Tuple2<String, Long>> tuple2s = new ArrayList<>();

            for (String a : stringLongHashMap.keySet()) {
                tuple2s.add(Tuple2.of(a,stringLongHashMap.get(a)));
            }


            return tuple2s;
        }

        @Override
        public HashMap<String, Long> merge(HashMap<String, Long> stringLongHashMap, HashMap<String, Long> acc1) {
            return null;
        }
    }

    private static class UrlViewCountResult extends ProcessAllWindowFunction<ArrayList<Tuple2<String,Long>>,UrlViewCount,TimeWindow> {


        @Override
        public void process(Context context, Iterable<ArrayList<Tuple2<String, Long>>> iterable, Collector<UrlViewCount> collector) throws Exception {
            ArrayList<Tuple2<String, Long>> next = iterable.iterator().next();

            next.sort(new Comparator<Tuple2<String, Long>>() {
                @Override
                public int compare(Tuple2<String, Long> o1, Tuple2<String, Long> o2) {
                    return (int)(o2.f1-o1.f1) ;
                }
            });

            collector.collect(new UrlViewCount(next.get(0).f0,next.get(0).f1,context.window().getStart(),context.window().getEnd()));
            collector.collect(new UrlViewCount(next.get(1).f0,next.get(1).f1,context.window().getStart(),context.window().getEnd()));
        }
    }
}

